GH-46421: [C++][Acero] Asofjoin respect PauseProducing from downstream. Version 2 - #51094

Rationale for this change

This is a second attempt at solving #46421. The first attempt in #46140 caused a performance regression because output batches merged multiple input batches and could grow very large. This repeatedly grew, reallocated, and copied the unmaterialized table. This PR instead redesigns the node's data flow and addresses the related correctness and execution issues together. The new candidate model also supports an inclusive tolerance range, allowing nearest-neighbor joins across both directions and exact-match exclusion without another special-case option.

What changes are included in this PR?

• Execution and materialization model: Replace the dedicated processing thread with a coordinator and independently scheduled processing for each RHS input. The coordinator activates one LHS batch at a time, while each RHS input finds matches and materializes its payload columns independently, implementing the parallel execution proposed in [C++] Parallel asof join node #34135. Once every RHS input completes, the coordinator reuses the original LHS arrays and scalars, resolving the unnecessary copying described in [C++] Asof-joins inefficiently copy the left hand side  #41873, appends the RHS columns, and emits one output batch preserving the LHS boundary and index fixing [C++] Make AsofJoinNode robust to input batch scheduling #36651. This also avoids the cross-boundary output growth amplified by the earlier asof_join_pause attempt. The same implementation supports threaded, serial, and ARROW_ENABLE_THREADING=OFF execution. Different backpressure semantics replaced old backpressure-controller construction should also eliminate the C++23 incomplete-type build failure reported in [C++][Acero] Build failure on OS X and cppstd 23 #47576.

• Backpressure: Bound the LHS and each RHS input queue using batch-count watermarks, applying pause and resume upstream as queues cross their watermarks. A downstream pause allows the active LHS batch to finish but prevents another from starting, fixing [C++][Acero] Asofjoin does not propagate pause upstream #46421.

• Key matching: Use hashes for lookup followed by exact by-key comparison, preventing collisions from producing incorrect joins and fixing [C++] AsofJoinNode 128-bit hashing #32894 without switching to 128-bit hashes. Support additional flat key types and scalar by keys. Benchmarks show no regression. Dictionary-encoded key columns remain unsupported.

• Non-key fields: Reuse LHS payloads directly and materialize RHS payloads using generic Arrow builders instead of the key encoder. This adds support for nested and dictionary-encoded payloads, including fixed-size lists, and addresses [C++][Acero] Not support type like Fixed Size List for non-key column in asof join node #44729.

• on-key normalization: Preserve the ordering of signed integer and temporal keys across zero, fixing [C++][Acero] Negative values in NormalizeTime are mapped to the highest uint64_t values #45876.

• Null on keys: Respect validity bitmaps so null LHS timestamps remain unmatched and null RHS timestamps are never selected, fixing [C++][Acero][Python] Asof join does not detect null times #46780.

• Tolerance ranges and tie-breaking: Add an AsofJoinNodeOptions constructor taking an inclusive [lower, upper] tolerance range and a prefer_earlier_on_tie rule that defaults to true. A right row is eligible when right.on - left.on is in the configured range; the closest eligible row is selected, with the preference resolving equal-distance matches on opposite sides. The existing scalar constructor maps to the same backward, forward, or exact interval as before. Ranges such as [-10, -1] exclude exact backward matches, addressing Add option to disable exact matches optional in join_asof #41786 without a separate Boolean option. Entirely non-positive ranges retain one latest candidate per by key, while ranges extending into the future retain ordered candidates and use binary search.

• Python bindings: Expose tolerance ranges and prefer_earlier_on_tie through pyarrow.acero.AsofJoinNodeOptions, Table.join_asof, and Dataset.join_asof. Python callers may continue passing an integer tolerance or pass a two-element (lower, upper) range.

• Input ordering: Validate that each input provides ordering compatible with its on key and preserve the LHS ordering in the output.

• Benchmarks: Expand coverage across threading modes, string-key sizes, tolerance directions, and input densities. Local results are approximately 2–109× faster than main on my machine.

• Tests: Re-enable and stabilize the generated-batch backpressure test, fixing [C++] Fix and re-enable Asof Join Backpresure test flakiness #36248. Add regression coverage for jittered input sequencing under threaded and serial execution; input-queue and downstream backpressure; completing or stopping an active LHS batch while paused; preserving LHS boundaries, indices, arrays, scalars, and ordering; exact collision arbitration; additional key and payload types; mixed scalar and array by keys; signed and null on keys; input-ordering validation; backward and forward exact-match exclusion; nearest-candidate tie-breaking; invalid ranges; normalized-key limits; and the public Python Table and Dataset paths.

Are these changes tested?

Yes. All 170 AsofJoin tests pass in both threaded and ARROW_ENABLE_THREADING=OFF builds. The formerly flaky batch-backpressure test also passes multiple repeated runs in each configuration. A freshly built PyArrow passes all 14 Table and Dataset join_asof tests, including the new range, tie-breaking, exact-exclusion, and null-time regressions.

Are there any user-facing changes?

• Additional flat by-key types are supported, including Boolean, fixed-size binary, and decimal types. Non-key payload columns are no longer restricted to types supported by the key encoder, enabling nested and dictionary-encoded payloads. Dictionary-encoded by keys remain unsupported.

• If an LHS ExecBatch contains a scalar representing a constant column, AsofJoin preserves it as a scalar in the output instead of materializing it as an array.

• AsofJoin emits exactly one output batch for each LHS input batch, preserving its length, index, and boundary. This includes zero-length LHS batches.

• With threaded execution, matching and materialization for separate RHS inputs may run concurrently. With use_threads=false or ARROW_ENABLE_THREADING=OFF, the same implementation runs serially without creating a dedicated processing thread.

• AsofJoin now participates in backpressure. The number of queued batches is bounded per input, and downstream pause prevents new LHS batches from starting after the active batch completes.

• Hash collisions can no longer cause different by keys to match because hash matches are verified using exact key equality.

• Signed on keys are ordered correctly across zero. Null LHS on values remain unmatched, and null RHS on values are never selected.

• The tolerance may be an inclusive lower/upper range. Ranges wholly in the past or future retain the expected directional behavior, while ranges spanning zero select the nearest eligible row. Equal-distance matches prefer the earlier row by default and may be configured to prefer the later row.

• PyArrow Table.join_asof and Dataset.join_asof accept either the existing integer tolerance or a two-element tolerance range, plus the optional prefer_earlier_on_tie argument.

• Plans with unordered or incompatible input ordering are now rejected during construction.

• GitHub Issue: [C++][Acero] Asofjoin does not propagate pause upstream #46421